package com.sunday.common.mq.rocket.brave.study.base.e4_Batch_Message;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 4 批量消息样例
 * 批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic，
 * 相同的waitStoreMsgOK，而且不能是延时消息。此外，这一批消息的总大小不应超过4MB。
 */
@Slf4j
public class BatchMessage {

    String topic = "BatchTest";

    /**
     * 4.1 发送批量消息
     * 如果您每次只发送不超过4MB的消息，则很容易使用批处理，样例如下：
     */
    @Test
    public void batchProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("batch_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();


        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));

        // 发送消息到一个Broker
        SendResult sendResult = producer.send(messages);
        // 通过sendResult返回消息是否成功送达
        log.info("{}", sendResult);

        producer.shutdown();

    }

    /**
     * 4.2 消息列表分割
     * 复杂度只有当你发送大批量时才会增长，你可能不确定它是否超过了大小限制（4MB）。这时候你最好把你的消息列表分割一下：
     */
    @Test
    public void batchSplitterProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("batch_producer_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        // 启动Producer实例
        producer.start();


        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, "TagA", "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, "TagA", "OrderID003", "Hello world 2".getBytes()));

        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                SendResult sendResult = producer.send(listItem);
                log.info("{}", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                //处理error
            }
        }

        producer.shutdown();

    }

    @AfterEach
    public void consumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch_consumer_group_name");

        // 设置NameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");

        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe(topic, "*");

        /**
         * https://github.com/apache/rocketmq/blob/master/docs/cn/best_practice.md
         *
         * 简单来说这个就是并行线程
         *
         * 2.2 消费速度慢的处理方式
         * 1 提高消费并行度
         * 绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，
         * 可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：
         *
         * 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。
         * 可以通过加机器，或者在已有机器启动多个进程的方式。
         * 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax实现。
         */
//        consumer.setConsumeThreadMin(1); // default 20
//        consumer.setConsumeThreadMax(1); // default 20

        /**
         * 这个就是一次批量获取的消息数量
         *
         * 2 批量方式消费
         * 某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，
         * 这样即可大幅度提高消费的吞吐量，
         * 通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。
         *
         * 也就是 consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) 中的 msgs 为什么是一个list
         */
//        consumer.setConsumeMessageBatchMaxSize(10); // default 1

        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            log.info("Receive New Messages: {}", msgs.size());
            msgs.stream().forEach(messageExt -> log.info("{}", messageExt));
            // 标记该消息已经被成功消费
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动消费者实例
        consumer.start();
        log.info("Consumer Started.");

        TimeUnit.SECONDS.sleep(10);

    }

}
